package com.gome.han.bigdata.spark.core.rdd.operation.transformation

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * @author Hanpeng
 * @date 2021/1/12 21:38
 * @description:
 *              只有两条输出
 *
 */
object MapPartitionsOperation {
  def main(args: Array[String]): Unit = {
    //TODO 准备环境
    val sparkConf:SparkConf = new SparkConf();
    sparkConf.setMaster("local[*]").setAppName("operationMap");
    val sparkContext:SparkContext = new SparkContext(sparkConf);
    //TODO 创建RDD
    val seq: Seq[Int] = Seq[Int](1, 2, 3, 4);
    val rdd: RDD[Int] = sparkContext.makeRDD(seq,2);
    //以分区为单位进行数据转换操作 提高了效率
    //但是会将整个分区的数据加载到内存中进行引用
    //处理完的数据不会立即释放,存在对象引用 多以在内存较小 数据量较大的情况下 容易出现内存溢出。
    //
    val mapRdd: RDD[Int] = rdd.mapPartitions(
      ite => {
        println(">>>>>>\n")
        ite.map(num=> num*2);
      }
    )

    mapRdd.collect();
    // TODO 关闭环境
    sparkContext.stop()
  }
}
